//package com.ruoyi.common.kafka;
//
//import org.apache.kafka.clients.consumer.*;
//import org.apache.kafka.common.TopicPartition;
//import java.time.Duration;
//import java.util.*;
//
///**
// * @Author: 倚天照海
// */
//public class MyConsumer {
//
//    public void consume() {
//        MyKafkaConfig kafkaConfig = new MyKafkaConfig();
//        Properties properties = new Properties();
//        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfig.getBootstrapServer());
//        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, kafkaConfig.getKeyDeserializerClass());
//        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, kafkaConfig.getValueDeserializerClass());
//        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, kafkaConfig.getConsumerGroupId());
//        /**
//         * ConsumerConfig.AUTO_OFFSET_RESET_CONFIG表示当kafka中未初始化offset或当前offset不存在时，消费者自动重置offset的方式，默认是latest。
//         * What to do when there is no initial offset in Kafka
//         * or if the current offset does not exist any more on the server (e.g. because that data has been deleted):
//         * <ul>
//         *     <li>earliest: automatically reset the offset to the earliest offset</li>
//         *     <li>latest: automatically reset the offset to the latest offset</li>
//         *     <li>none: throw exception to the consumer if no previous offset is found for the consumer's group</li>
//         *     <li>anything else: throw exception to the consumer.</li>
//         * </ul>
//         */
//        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
//        // 是否开启自动提交，默认开启。自动提交是异步提交，开启自动提交可能会造成数据丢失或重复消费数据
//        // properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
//        // 自动提交的间隔时间(多长时间会触发自动提交)，默认是5秒
//        // properties.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "10000");
//        // kafka的消费者是按批次拉取数据，该参数是设置一批最多拉取多少条数据
//        // properties.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1000");
//
//        List<String> topics = Collections.singletonList(kafkaConfig.getTopic());
//        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
//        doConsume(consumer, topics);
//    }
//
//    private void doConsume(KafkaConsumer<String, String> consumer, List<String> topics) {
//        // 消费者订阅主题消息，多个consumer会动态负载均衡多个分区
//        // 例如有两个分区，最开始只启动一个consumer，会给这个consumer分配两个分区，它会消费两个分区的数据。
//        // 然后在同一个消费者组内又启动一个consumer，此时会把第一个consumer的两个分区都撤销，再随机给这两个consumer分别分配一个分区
//        consumer.subscribe(topics, new ConsumerRebalanceListener() {
//            @Override
//            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
//                System.out.println("-----------------onPartitionsRevoked撤销的分区是:---------------");
//                Iterator<TopicPartition> iterator = partitions.iterator();
//                while (iterator.hasNext()) {
//                    TopicPartition next = iterator.next();
//                    System.out.println(next.partition());
//                }
//            }
//
//            @Override
//            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
//                System.out.println("-----------------onPartitionsAssigned分配的分区是:---------------");
//                Iterator<TopicPartition> iterator = partitions.iterator();
//                while (iterator.hasNext()) {
//                    TopicPartition next = iterator.next();
//                    System.out.println(next.partition());
//                }
//            }
//        });
//        while (true) {
//            // 消费者拉取消息，设置等待时间，按批次拉取，一批拉取的数据是0-n条，每次poll是同时拉取多个分区的数据
//            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(0));
//            while (!consumerRecords.isEmpty()) {
//                // 每次拉取数据的条数
//                System.out.println("----------------consumerRecords.count------------------" + consumerRecords.count());
//                // 方式一：按分区分别处理每个分区的数据
//                Set<TopicPartition> partitions = consumerRecords.partitions();
//                // kafka中consumer是按照分区粒度提交维护offset的，将offset提交到__consumer_offsets中
//                // 如果关闭自动提交，使用手动提交offset，则有三种粒度同步提交offset：
//                // 1.按每条消息粒度同步提交offset
//                // 2.按每个分区粒度同步提交offset
//                // 3.按poll的一批数据粒度同步提交offset
////                for (TopicPartition topicPartition : partitions) {
////                    // 分别获取每个分区的数据记录，且分区内的数据是有序的，可以用多线程并行处理每个分区的数据
////                    List<ConsumerRecord<String, String>> partitionRecords = consumerRecords.records(topicPartition);
////                    for (ConsumerRecord<String, String> record : partitionRecords) {
////                        // 一个消费者可以消费多个分区，一个分区只能被同一个消费者组中的一个消费者消费
////                        int partition = record.partition();
////                        long offset = record.offset();
////                        // TODO 获取数据处理复杂的业务逻辑，最终将数据持久化到数据库中
////                        System.out.println("key=" + record.key() + ", value=" + record.value()
////                                + ", partition=" + partition + ", offset=" + offset);
////
////                        // 粒度1.按每条消息粒度同步提交offset
////                        /*OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(offset);
////                        Map<TopicPartition, OffsetAndMetadata> map = new HashMap<>();
////                        map.putIfAbsent(topicPartition, offsetAndMetadata);
////                        consumer.commitSync(map);*/
////                    }
////                    // 粒度2.按每个分区粒度同步提交offset
////                    // 获取分区最后一条消息记录的offset，将offset提交到__consumer_offsets中
////                    long offset = partitionRecords.get(partitionRecords.size() - 1).offset();
////                    OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(offset);
////                    Map<TopicPartition, OffsetAndMetadata> map = new HashMap<>();
////                    map.putIfAbsent(topicPartition, offsetAndMetadata);
////                    consumer.commitSync(map);
////                }
//                // 粒度3.按poll的一批数据粒度同步提交offset
//                // consumer.commitSync();
//
//                // 方式二：不区分分区，将多个分区的数据放在一起逐条处理
//                for (ConsumerRecord<String, String> record : consumerRecords) {
//                    // 一个消费者可以消费多个分区，一个分区只能被同一个消费者组中的一个消费者消费
//                    int partition = record.partition();
//                    long offset = record.offset();
//                    // TODO 获取数据处理复杂的业务逻辑，最终将数据持久化到数据库中
//                    System.out.println("key=" + record.key() + ", value=" + record.value()
//                            + ", partition=" + partition + ", offset=" + offset);
//                }
//                consumer.commitSync();
//            }
//        }
//    }
//}
